package com.pai4j.user.service.relation;
import com.pai4j.common.enums.RedisKeyEnum;
import com.pai4j.common.bean.PAIPageResponseBeanUtil;
import com.pai4j.common.constants.WebConstant;
import com.pai4j.common.enums.AccountRelationStatusEnum;
import com.pai4j.common.enums.AccountRelationTypeEnum;
import com.pai4j.common.enums.RoleCodeEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.DateUtils;
import com.pai4j.common.util.RedisUtil;
import com.pai4j.common.util.ThreadPoolExecutorUtil;
import com.pai4j.domain.vo.request.mess.ChatMessageRequestVO;
import com.pai4j.domain.vo.response.UserBaseResponseInfoVO;
import com.pai4j.user.repository.dao.IAccountRelationDAO;
import com.pai4j.user.repository.entity.AccountRelationEntity;
import com.pai4j.user.service.UserService;
import jakarta.transaction.Transactional;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

public abstract class AbstractAccountRelationService implements AccountRelationValidator, AccountRelationWrapper {
    
    private static final Log log = LogFactory.getLog(AbstractAccountRelationService.class);
    @Autowired
    protected IAccountRelationDAO accountRelationDAO;

    @Autowired
    UserService userService;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;

    @Autowired
    protected com.pai4j.user.service.approval.PendingApprovalService pendingApprovalService;

    private UserBaseResponseInfoVO sourceAccountInfo;

    private UserBaseResponseInfoVO targetAccountInfo;

    protected void init(String sourceAccount, String targetAccount) {
        this.sourceAccountInfo = userService.getUserBaseInfoByUserId(sourceAccount);
        this.targetAccountInfo = userService.getUserBaseInfoByUserId(targetAccount);
    }

    /**
     * 订阅关系类型，需由具体实现类返回
     *
     *
     * @return
     */
    protected abstract AccountRelationTypeEnum getRelationType();

    /**
     * 获取订阅状态
     *
     * @return
     */
    protected AccountRelationStatusEnum getSubscribeStatus() {
        return AccountRelationStatusEnum.CONFIRMED;
    }

    /**
     * 是否需要异步执行后置处理流程
     *
     * @return
     */
    protected abstract boolean isAsyncDoAfterComplete();

    /**
     * 完成核心流程后的业务处理，由各实现类自己实现
     * 比如：
     * 1、基于不同关系的缓存构建、刷新
     */
    protected abstract void doAfterComplete(Long id, String sourceAccount, String targetAccount, boolean subscribe);

    /**
     * 完成核心流程后的业务处理，由各实现类自己实现
     * 比如：
     * 1、基于不同关系的缓存构建、刷新
     */
    protected void execAfterComplete(Long id, String sourceAccount, String targetAccount, boolean subscribe) {
        /**
         * 后置流程公共逻辑实现
         */
        if (!subscribe) {
            this.removeTargetAccountRelationListCache(sourceAccount, targetAccount);
        }
        
        // 刷新粉丝数和关注数缓存
        this.refreshFollowersAndFollowingCountCache(sourceAccount, targetAccount);
        
        /**
         * 差异化逻辑各个实现类自己支持
         */
        this.doAfterComplete(id, sourceAccount, targetAccount, subscribe);
    }

    /**
     * 异步执行后置业务处理流程
     *
     * @param id
     * @param sourceAccount
     * @param targetAccount
     */
    protected void asyncDoAfterComplete(Long id, String sourceAccount, String targetAccount, boolean subscribe) {

        ThreadPoolExecutorUtil.execute(() -> execAfterComplete(id, sourceAccount, targetAccount, subscribe));
    }

    private void doExecAfterComplete(Long id, String sourceAccount, String targetAccount, boolean subscribe) {
        if (isAsyncDoAfterComplete()) {
            this.asyncDoAfterComplete(id, sourceAccount, targetAccount, subscribe);
        } else {
            this.execAfterComplete(id, sourceAccount, targetAccount, subscribe);
        }
    }

    /**
     * 前置基础校验
     *
     * @param sourceAccount
     * @param targetAccount
     */
    @Override
    public void validateAccountRequest(String sourceAccount, String targetAccount) {
        Assert.notNull(sourceAccountInfo, "source account is null!");
        Assert.isTrue(RoleCodeEnum.isAddRole(sourceAccountInfo.getRoleCode() ), "source account role is not support!");
        Assert.notNull(targetAccountInfo, "target account is null!");
        Assert.isTrue(!RoleCodeEnum.BLACK.getCode().equals(targetAccountInfo.getRoleCode()), "target account is black list user!");
    }

    /**
     * 关系订阅接口
     *
     * @param sourceAccount
     * @param targetAccount
     * @return
     */
    @Transactional
    @Override
    public Boolean subscribe(String sourceAccount, String targetAccount) {
        return this.subscribe(sourceAccount, targetAccount, false);
    }

    /**
     * 关系订阅接口（支持是否需要确认参数）
     *
     * @param sourceAccount 源账号
     * @param targetAccount 目标账号
     * @param needConfirmed 是否需要确认，false=普通关注（直接生效），true=好友申请（需要确认）
     * @return
     */
    @Transactional
    public Boolean subscribe(String sourceAccount, String targetAccount, boolean needConfirmed) {
        return this.subscribe(sourceAccount, targetAccount, needConfirmed, null);
    }

    /**
     * 关系订阅接口（支持推荐人参数）
     *
     * @param sourceAccount 源账号
     * @param targetAccount 目标账号
     * @param needConfirmed 是否需要确认，false=普通关注（直接生效），true=好友申请（需要确认）
     * @param recommenderAccount 推荐人账号（可选，用于推荐卡片场景）
     * @return
     */
    @Transactional
    public Boolean subscribe(String sourceAccount, String targetAccount, boolean needConfirmed, String recommenderAccount) {
        this.init(sourceAccount, targetAccount);
        /**
         * 前置参数校验
         */
        this.validateAccountRequest(sourceAccount, targetAccount);
        /**
         * 核心订阅流程处理
         */
        Long id = this.doSubscribe(sourceAccount, targetAccount, needConfirmed, recommenderAccount);
        /**
         * 执行后置流程
         */
        this.doExecAfterComplete(id, sourceAccount, targetAccount, Boolean.TRUE);
        return true;
    }

    /**
     * 执行取消订阅的持久化更新逻辑
     *
     * @param sourceAccount
     * @param targetAccount
     * @return
     */
    @Transactional
    @Override
    public Boolean unsubscribe(String sourceAccount, String targetAccount) {
        this.init(sourceAccount, targetAccount);
        /**
         * 前置参数校验
         */
        this.validateAccountRequest(sourceAccount, targetAccount);
        /**
         * 核心订阅流程处理
         */
        Long id = this.doUnSubscribe(sourceAccount, targetAccount);
        /**
         * 执行后置流程
         */
        this.doExecAfterComplete(id, sourceAccount, targetAccount, Boolean.FALSE);
        return true;
    }

    /**
     * 查询账号关系列表，走redis
     *
     * @param account
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Override
    public PAIPageResponseBeanUtil<UserBaseResponseInfoVO> listRelations(String account, Integer pageNo, Integer pageSize) {

        Integer type = this.getRelationType().getType();
        String prefix = "SAFE";
        String cacheKey = this.getRelationListRedisKey(account, type);
        String safeCacheKey = this.getRelationListRedisKey(prefix, account, type);
        if (!redisUtil.exists(cacheKey) && !redisUtil.exists(safeCacheKey)) {
            // 失效/没有好友关系, 刷新账号管理缓存列表
            this.initAccountRelationListCache(account, type);
            redisUtil.set(safeCacheKey, WebConstant.CACHE_NONE, 5 * 60);
        }
        Long total = redisUtil.zsize(cacheKey);
        List<UserBaseResponseInfoVO> relactionAccountList = new ArrayList<>();
        if (total > 0) {
            Set<String> relationCaches = redisUtil.zreverseRange(cacheKey, (pageNo - 1) * pageSize, pageNo * pageSize);
            relactionAccountList = userService.batchGetUserListInfoFromCache(relationCaches);
        }
        return PAIPageResponseBeanUtil.success(pageNo, pageSize, total, relactionAccountList);
    }

    /**
     * 查询账号下的所有订阅用户列表
     *
     * @param account
     * @return
     */
    @Override
    public Set<String> getAllSubscriber(String account) {
        Integer type = this.getRelationType().getType();
        String prefix = "SAFE";
        String cacheKey = this.getRelationListRedisKey(account, type);
        String safeCacheKey = this.getRelationListRedisKey(prefix, account, type);
        if (!redisUtil.exists(cacheKey) && !redisUtil.exists(safeCacheKey)) {
            // 失效/没有好友关系, 刷新账号管理缓存列表
            this.initAccountRelationListCache(account, type);
            redisUtil.set(safeCacheKey, WebConstant.CACHE_NONE, 5 * 60);
        }
        return redisUtil.zreverseRange(cacheKey, 0, Integer.MAX_VALUE);
    }

    /**
     * 获取用户的粉丝数量
     * 粉丝是指关注该用户的其他用户数量（该用户作为target_account的关系数量）
     *
     * @param account 用户账号
     * @return 粉丝数量
     */
    public Long getFollowersCount(String account) {
        Integer type = this.getRelationType().getType();
        String countCacheKey = this.getFollowersCountRedisKey(account, type);
        
        // 先从缓存获取
        if (redisUtil.exists(countCacheKey)) {
            String countStr = redisUtil.get(countCacheKey);
            if (countStr != null && !WebConstant.CACHE_NONE.equals(countStr)) {
                return Long.valueOf(countStr);
            }
        }
        
        // 缓存不存在或失效，查询数据库
        Long followersCount = accountRelationDAO.countFollowers(account, type, AccountRelationStatusEnum.CONFIRMED.getStatus());
        
        // 更新缓存，设置5分钟过期时间
        redisUtil.set(countCacheKey, String.valueOf(followersCount), 5 * 60);
        
        return followersCount;
    }

    /**
     * 获取用户的关注数量
     * 关注是指该用户关注的其他用户数量（该用户作为source_account的关系数量）
     *
     * @param account 用户账号
     * @return 关注数量
     */
    public Long getFollowingCount(String account) {
        Integer type = this.getRelationType().getType();
        String countCacheKey = this.getFollowingCountRedisKey(account, type);
        
        // 先从缓存获取
        if (redisUtil.exists(countCacheKey)) {
            String countStr = redisUtil.get(countCacheKey);
            if (countStr != null && !WebConstant.CACHE_NONE.equals(countStr)) {
                return Long.valueOf(countStr);
            }
        }
        
        // 缓存不存在或失效，查询数据库
        Long followingCount = accountRelationDAO.countFollowing(account, type, AccountRelationStatusEnum.CONFIRMED.getStatus());
        
        // 更新缓存，设置5分钟过期时间
        redisUtil.set(countCacheKey, String.valueOf(followingCount), 5 * 60);
        
        return followingCount;
    }

    @Override
    public boolean isSubscribe(String userAccount, String account) {
        Integer type = this.getRelationType().getType();
        String prefix = "SAFE";
        String cacheKey = this.getRelationListRedisKey(account, type);
        String safeCacheKey = this.getRelationListRedisKey(prefix, account, type);
        if (!redisUtil.exists(cacheKey) && !redisUtil.exists(safeCacheKey)) {
            // 失效/没有好友关系, 刷新账号管理缓存列表
            this.initAccountRelationListCache(account, type);
            redisUtil.set(safeCacheKey, WebConstant.CACHE_NONE, 5 * 60);
        }
        return redisUtil.zIsExist(cacheKey, userAccount);
    }

    /**
     * 添加targetAccount到sourceAccount的好友列表缓存
     *
     * @param sourceAccount
     * @param targetAccount
     */
    protected void addTargetAccountRelationListCache(String sourceAccount, String targetAccount) {
        // 获取好友列表redis key
        AccountRelationTypeEnum relationType = getRelationType();
        String cacheKey = this.getRelationListRedisKey(sourceAccount, relationType.getType());
        if (!redisUtil.exists(cacheKey)) {
            // 第一次订阅或者缓存失效
            // 查询数据库，主动刷新redis
            this.initAccountRelationListCache(sourceAccount, relationType.getType());
            redisUtil.zadd(cacheKey, targetAccount, DateUtils.getCurrentTimeMillis(),
                    RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getExpireTime());
        }
        /**
         * 添加targetAccount到sourceAccount的好友列表缓存，当前时间作为sorted set 的score，后续可用于倒序查询
         */
        redisUtil.zadd(cacheKey, targetAccount, DateUtils.getCurrentTimeMillis(),
                RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getExpireTime());
    }

    /**
     * 删除用户关系账号
     *
     * @param account
     * @param removeAccount
     */
    protected void removeTargetAccountRelationListCache(String account, String removeAccount) {
        // 获取好友列表redis key
        AccountRelationTypeEnum relationType = AccountRelationTypeEnum.getByTargetAccountRoleCode(
                targetAccountInfo.getRoleCode());
        String cacheKey = this.getRelationListRedisKey(account, relationType.getType());
        redisUtil.zremove(cacheKey, removeAccount);
    }

    /**
     * 初始化主动刷新账号关系列表缓存
     *
     * @param account
     * @param type
     */
    public void initAccountRelationListCache(String account, Integer type) {
        List<AccountRelationEntity> accountRelationEntities =
                this.getAccountRelationList(account, type);
        if (CollectionUtils.isEmpty(accountRelationEntities)) {
           return;
        }
        String cacheKey = this.getRelationListRedisKey(account, type);
        Set<ZSetOperations.TypedTuple<String>> tuples = accountRelationEntities.stream().
                map(ar -> {
                    String value = ar.getSourceAccount().equals(account) ? ar.getTargetAccount() : ar.getSourceAccount();
                    Double score = Double.valueOf(ar.getUpdateDate().getTime());
                    return new DefaultTypedTuple<>(value, score);
                }).collect(Collectors.toSet());
        redisUtil.zadd(cacheKey, tuples, RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getExpireTime());
    }

    private String getRelationListRedisKey(String sourceAccount, Integer relationType) {
        String prefix = "DATA";
        return this.getRelationListRedisKey(prefix, sourceAccount, relationType);
    }

    private String getRelationListRedisKey(String prefix, String sourceAccount, Integer relationType) {
        String typePrefix = "T";
        String accountPrefix = "AC";
        return RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getKey(
                prefix,
                typePrefix,
                String.valueOf(relationType),
                accountPrefix,
                sourceAccount);
    }

    /**
     * 获取粉丝数量缓存key
     *
     * @param account 用户账号
     * @param relationType 关系类型
     * @return Redis缓存key
     */
    private String getFollowersCountRedisKey(String account, Integer relationType) {
        String prefix = "FOLLOWERS_COUNT";
        String typePrefix = "T";
        String accountPrefix = "AC";
        return RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getKey(
                prefix,
                typePrefix,
                String.valueOf(relationType),
                accountPrefix,
                account);
    }

    /**
     * 获取关注数量缓存key
     *
     * @param account 用户账号
     * @param relationType 关系类型
     * @return Redis缓存key
     */
    private String getFollowingCountRedisKey(String account, Integer relationType) {
        String prefix = "FOLLOWING_COUNT";
        String typePrefix = "T";
        String accountPrefix = "AC";
        return RedisKeyEnum.ACCOUNT_RELATION_LIST_CACHE.getKey(
                prefix,
                typePrefix,
                String.valueOf(relationType),
                accountPrefix,
                account);
    }

    /**
     * 刷新粉丝数和关注数缓存
     * 当关系发生变化时，需要同时刷新sourceAccount的关注数和targetAccount的粉丝数
     *
     * @param sourceAccount 源账号
     * @param targetAccount 目标账号
     */
    protected void refreshFollowersAndFollowingCountCache(String sourceAccount, String targetAccount) {
        Integer type = this.getRelationType().getType();
        
        // 刷新sourceAccount的关注数缓存
        String followingCountCacheKey = this.getFollowingCountRedisKey(sourceAccount, type);
        redisUtil.remove(followingCountCacheKey);
        
        // 刷新targetAccount的粉丝数缓存
        String followersCountCacheKey = this.getFollowersCountRedisKey(targetAccount, type);
        redisUtil.remove(followersCountCacheKey);
    }


    /**
     * 根据关系ID获取实体信息
     *
     * @param relationId
     * @return
     */
    public AccountRelationEntity getRelationEntityById(Long relationId) {

        return accountRelationDAO.findById(relationId).orElse(null);
    }

    /**
     * 更新关系实体信息
     *
     * @param updateAccountRelationEntity
     * @return
     */
    public AccountRelationEntity updateAccountRelationEntity(AccountRelationEntity updateAccountRelationEntity) {

        Assert.notNull(updateAccountRelationEntity, "entity is null!");
        return accountRelationDAO.save(updateAccountRelationEntity);
    }

    /**
     * 查询数据库，获取账号的好友/订阅列表
     *
     * @param account
     * @return
     */
    public List<AccountRelationEntity> getAccountRelationList(String account, Integer type) {

        return accountRelationDAO.getAccountRelations(account, type, AccountRelationStatusEnum.CONFIRMED.getStatus());
    }

    /**
     * 执行订阅的持久化更新逻辑（支持是否需要确认参数）
     *
     * @param sourceAccount 源账号
     * @param targetAccount 目标账号
     * @param needConfirmed 是否需要确认，false=普通关注（直接生效），true=好友申请（需要确认）
     * @param recommenderAccount 推荐人账号（可选）
     * @return
     */
    private Long doSubscribe(String sourceAccount, String targetAccount, boolean needConfirmed, String recommenderAccount) {
        AccountRelationEntity entity = this.getAccountRelationEntity(sourceAccount, targetAccount);
        Assert.isTrue(entity == null || !AccountRelationStatusEnum.isConfirmed(entity.getStatus()), "重复订阅！");
        
        // 根据needConfirmed参数决定关系状态
        AccountRelationStatusEnum targetStatus;
        if (needConfirmed) {
            // 需要确认的场景（如好友申请），使用子类重写的状态或默认等待确认状态
            targetStatus = this.getSubscribeStatus();
        } else {
            // 不需要确认的场景（如普通关注），直接设置为已确认状态
            targetStatus = AccountRelationStatusEnum.CONFIRMED;
        }
        
        if (entity != null) {
            Date now = new Date();
            entity.setStatus(targetStatus.getStatus());
            entity.setUpdateDate(now);
            entity.setCreateDate(now);
        } else {
            entity = buildAccountRelationEntity(sourceAccount, targetAccount, targetStatus);
        }
        Long relationId = this.saveAccountRelation2DB(entity);
        
        // 【强一致性】如果是待审核的好友申请，同步创建统一待审核记录
        if (needConfirmed && AccountRelationStatusEnum.WAIT_CONFIRMED.equals(targetStatus) 
                && com.pai4j.common.enums.AccountRelationTypeEnum.USER_USER.equals(this.getRelationType())) {
            try {
                pendingApprovalService.createPendingApproval(
                        com.pai4j.common.enums.PendingApprovalTypeEnum.FRIEND_REQUEST.getType(),
                        relationId,
                        sourceAccount,  // 申请人
                        targetAccount,  // 审批人
                        targetAccount,  // targetId（好友申请的目标ID就是targetAccount）
                        null,  // 申请留言（可从entity扩展字段获取，这里简化处理）
                        targetStatus.getStatus(),
                        null,  // 过期时间（可扩展）
                        recommenderAccount  // 推荐人账号
                );
                log.info("好友申请同步创建统一待审核记录成功：relationId={}, recommender={}");
            } catch (Exception e) {
                log.error("好友申请同步创建统一待审核记录失败，事务将回滚：relationId={}");
                throw new RuntimeException("创建统一待审核记录失败", e);
            }
        }
        
        return relationId;
    }

    /**
     * 执行取消订阅的持久化更新逻辑
     * @param sourceAccount
     * @param targetAccount
     * @return
     */
    private Long doUnSubscribe(String sourceAccount, String targetAccount) {
        AccountRelationEntity entity = this.getAccountRelationEntity(sourceAccount, targetAccount);
        Assert.isTrue(entity != null && AccountRelationStatusEnum.isConfirmed(entity.getStatus()), "未订阅，取消操作失败！");
        entity.setStatus(AccountRelationStatusEnum.DELETED.getStatus());
        entity.setUpdateDate(new Date());
        return this.saveAccountRelation2DB(entity);
    }

    private AccountRelationEntity getAccountRelationEntity(String sourceAccount, String targetAccount) {

        return accountRelationDAO.findAllBySourceAccountAndTargetAccount(sourceAccount, targetAccount);
    }

    private Long saveAccountRelation2DB(AccountRelationEntity entity) {

        return accountRelationDAO.save(entity).getId();
    }

    private AccountRelationEntity buildAccountRelationEntity(String sourceAccount, String targetAccount,
                                                             AccountRelationStatusEnum statusEnum) {
        AccountRelationEntity entity = new AccountRelationEntity();
        entity.setSourceAccount(sourceAccount);
        entity.setTargetAccount(targetAccount);
        entity.setStatus(statusEnum.getStatus());
        entity.setType(getRelationType().getType());
        entity.setCreateDate(new Date());
        entity.setUpdateDate(new Date());
        return entity;
    }

    /**
     * 一旦好友添加成功/公众号订阅成功等将 targetAccount添加到sourceAccount的最新对话列表缓存
     *
     * @param sourceAccount
     * @param targetAccount
     */
    protected void addTargetAccount2SourceLastChattingList(String sourceAccount, String targetAccount) {

        this.addTargetAccount2SourceLastChattingList(sourceAccount, targetAccount, System.currentTimeMillis());
    }

    protected void addTargetAccount2SourceLastChattingList(String sourceAccount, String targetAccount, Long lastTime) {
        /**
         * 好友关系订阅成功，走MQ通知UGC服务，需要刷新对话列表缓存
         */
        ChatMessageRequestVO message = new ChatMessageRequestVO();
        message.setSenderId(sourceAccount);
        message.setReceiverId(targetAccount);
        message.setTime(lastTime);
        messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REFRESH, message);
    }

    /**
     * 退订/退群后，从最近会话列表移除目标账号
     *
     * @param sourceAccount 源账号
     * @param targetAccount 目标账号
     */
    protected void removeTargetAccountFromSourceLastChattingList(String sourceAccount, String targetAccount) {
        ChatMessageRequestVO message = new ChatMessageRequestVO();
        message.setSenderId(sourceAccount);
        message.setReceiverId(targetAccount);
        message.setTime(System.currentTimeMillis());
        messageQueueProducer.send(MessageQueueEnum.QUEUE_CHATTING_LIST_REMOVE, message);
    }
}
